package trident.wordcount;

import org.apache.storm.shade.org.apache.commons.collections.MapUtils;
import org.apache.storm.trident.operation.Aggregator;
import org.apache.storm.trident.operation.TridentCollector;
import org.apache.storm.trident.operation.TridentOperationContext;
import org.apache.storm.trident.tuple.TridentTuple;
import org.apache.storm.tuple.Values;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;

public class CountLocationMapAggregator implements Aggregator<Map<String, Integer>> {
    private final Logger logger = LoggerFactory.getLogger(CountLocationMapAggregator.class);
    private int partitionIndex;

    @Override
    public Map<String, Integer> init(Object batchId, TridentCollector collector) {
        return new HashMap<>();
    }

    @Override
    public void aggregate(Map<String, Integer> val, TridentTuple tuple, TridentCollector collector) {
        String location = tuple.getStringByField("location");
        // MapUtil.getIntValue() default value
        val.put(location, MapUtils.getIntValue(val, location, 0) + 1);
    }

    @Override
    public void complete(Map<String, Integer> val, TridentCollector collector) {
        logger.info("partition index"+partitionIndex);
        collector.emit(new Values(val));
    }

    @Override
    public void prepare(Map<String, Object> conf, TridentOperationContext context) {
        partitionIndex = context.getPartitionIndex();
    }

    @Override
    public void cleanup() {

    }
}
